package mapreduce.phase3;

import java.io.IOException;
import java.util.StringTokenizer;

import mapreduce.ColumnGrp;
import mapreduce.Point;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

/**
 * Recalculate vertices in specific columns by the help of new picture gained  from 
 * part 2
 */

@SuppressWarnings("deprecation")
public class Map3 extends MapReduceBase implements
	Mapper<LongWritable, Text, IntWritable, Text> {
		private final static IntWritable part = new IntWritable(1);

		public void map(LongWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
		throws IOException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			assert (tokenizer.hasMoreTokens());
			int colgrp = Integer.parseInt(tokenizer.nextToken());;
			if(colgrp>0)
			{
				assert (tokenizer.hasMoreTokens());
				Point p1 = Point.fromString(tokenizer.nextToken());
				assert (tokenizer.hasMoreTokens());
				Point p2 = Point.fromString(tokenizer.nextToken());
			
				Text t = new Text(p1.toString() + " " + p2.toString());
				part.set(ColumnGrp.getMainGroup(p1));
				output.collect(part, t);
				int secondryGroup = ColumnGrp.getSecondryGroup(p1);
				if(secondryGroup != -1) {
					part.set(secondryGroup);
					output.collect(part, t);
				}
			}
			else
			{
				assert (tokenizer.hasMoreTokens());
				Text t = new Text(tokenizer.nextToken());
				part.set(colgrp);
				output.collect(part, t);
			}
		}
	}
